package com.cfg.waf.kafka;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;
import java.util.Random;

public class Producer extends Thread{
    private Properties properties;
    private String topic;

    public Producer(Properties properties, String topic) {
        this.properties = properties;
        this.topic = topic;
    }

    public void run() {
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);

        try {
            while (true) {
                String msg = "Hello " + new Random().nextInt(100);
                ProducerRecord<String, String> record = new ProducerRecord<>(topic, msg);
                kafkaProducer.send(record);
                System.out.println("消息发送成功:" + msg);
                Thread.sleep(500);
            }
        }catch (Exception e) {
            e.printStackTrace();
        } finally {
            kafkaProducer.close();
        }
    }
}
